package com.example.mq2.mqclient;

import com.example.mq2.common.BinaryTool;
import com.example.mq2.common.MqException;
import com.example.mq2.mqserver.core.BasicProperties;
import com.example.mq2.mqserver.core.Consumer;
import com.example.mq2.mqserver.core.ExchangeType;
import com.example.mq2.network.*;

import javax.websocket.Decoder;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

public class Channel {

    private String channelId;
    //表示自己是来自于哪一个 Connection
    private Connection connection;
    //用来接收响应，key 是 rid
    private ConcurrentHashMap<String, BasicResponse> ridResponseMap = new ConcurrentHashMap<>();
    //一个 channel 一次通信只能管理一个回调
    private Consumer consumer;

    public Channel(String channelId, Connection connection) {
        this.channelId = channelId;
        this.connection = connection;
    }

    public String getChannelId() {
        return channelId;
    }

    public void setChannelId(String channelId) {
        this.channelId = channelId;
    }

    public Connection getConnection() {
        return connection;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    public ConcurrentHashMap<String, BasicResponse> getRidResponseMap() {
        return ridResponseMap;
    }

    public void setRidResponseMap(ConcurrentHashMap<String, BasicResponse> ridResponseMap) {
        this.ridResponseMap = ridResponseMap;
    }

    public Consumer getConsumer() {
        return consumer;
    }

    public void setConsumer(Consumer consumer) {
        this.consumer = consumer;
    }

    /**
     * 客户端接收到响应后放回到 map 中
     * @param basicResponse
     */
    public void putResponse(BasicResponse basicResponse) {
        synchronized (this) {
            ridResponseMap.put(basicResponse.getRid(), basicResponse);
            notifyAll();
        }
    }

    public boolean create(Channel channel) throws IOException {
        BasicRequest basicRequest = new BasicRequest();
        basicRequest.setRid(createRidByUUID());
        basicRequest.setChannelId(channelId);
        byte[] payload = BinaryTool.toBytes(basicRequest);

        Request request = new Request();
        request.setType(0x1);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(basicRequest.getRid());
        return basicResponse.isOk();
    }

    private BasicResponse waitResponse(String rid) {
        BasicResponse basicResponse = null;
        synchronized (this) {
            while((basicResponse = ridResponseMap.get(rid)) == null) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        return basicResponse;
    }

    private String createRidByUUID() {
        return "R-" + UUID.randomUUID();
    }

    public boolean close() throws IOException {
        BasicRequest basicRequest = new BasicRequest();
        basicRequest.setRid(createRidByUUID());
        basicRequest.setChannelId(channelId);
        byte[] payload = BinaryTool.toBytes(basicRequest);
        Request request = new Request();
        request.setType(0x2);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(basicRequest.getRid());
        return basicResponse.isOk();
    }

    public boolean exchangeDeclare(String name, ExchangeType type, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) throws IOException {
        ExchangeDeclareArguments exchangeDeclareArguments = new ExchangeDeclareArguments();
        exchangeDeclareArguments.setRid(createRidByUUID());
        exchangeDeclareArguments.setChannelId(channelId);
        exchangeDeclareArguments.setName(name);
        exchangeDeclareArguments.setType(type);
        exchangeDeclareArguments.setDurable(durable);
        exchangeDeclareArguments.setAutoDelete(autoDelete);
        exchangeDeclareArguments.setArguments(arguments);
        byte[] payload = BinaryTool.toBytes(exchangeDeclareArguments);
        Request request = new Request();
        request.setType(0x3);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(exchangeDeclareArguments.getRid());
        return basicResponse.isOk();
    }

    public boolean exchangeDelete(String name) throws IOException {
        ExchangeDeleteArguments arguments = new ExchangeDeleteArguments();
        arguments.setRid(createRidByUUID());
        arguments.setChannelId(channelId);
        arguments.setName(name);
        byte[] payload = BinaryTool.toBytes(arguments);
        Request request = new Request();
        request.setType(0x4);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(arguments.getRid());
        return basicResponse.isOk();
    }

    public boolean queueDeclare(String name, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) throws IOException {
        QueueDeclareArguments queueDeclareArguments = new QueueDeclareArguments();
        queueDeclareArguments.setRid(createRidByUUID());
        queueDeclareArguments.setChannelId(channelId);
        queueDeclareArguments.setName(name);
        queueDeclareArguments.setDurable(durable);
        queueDeclareArguments.setExclusive(exclusive);
        queueDeclareArguments.setAutoDelete(autoDelete);
        queueDeclareArguments.setArguments(arguments);
        byte[] payload = BinaryTool.toBytes(queueDeclareArguments);
        Request request = new Request();
        request.setType(0x5);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(queueDeclareArguments.getRid());
        return basicResponse.isOk();
    }

    public boolean queueDelete(String name) throws IOException {
        QueueDeleteArguments arguments = new QueueDeleteArguments();
        arguments.setRid(createRidByUUID());
        arguments.setChannelId(channelId);
        arguments.setName(name);
        byte[] payload = BinaryTool.toBytes(arguments);
        Request request = new Request();
        request.setType(0x6);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(arguments.getRid());
        return basicResponse.isOk();
    }

    public boolean queueBind(String exchangeName, String queueName, String bindingKey) throws IOException {
        QueueBindArguments arguments = new QueueBindArguments();
        arguments.setRid(createRidByUUID());
        arguments.setChannelId(channelId);
        arguments.setExchangeName(exchangeName);
        arguments.setQueueName(queueName);
        arguments.setBindingKey(bindingKey);
        byte[] payload = BinaryTool.toBytes(arguments);
        Request request = new Request();
        request.setType(0x7);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(arguments.getRid());
        return basicResponse.isOk();
    }

    public boolean queueUnBind(String exchangeName, String queueName) throws IOException {
        QueueUnBindingArguments arguments = new QueueUnBindingArguments();
        arguments.setRid(createRidByUUID());
        arguments.setChannelId(channelId);
        arguments.setExchangeName(exchangeName);
        arguments.setQueueName(queueName);
        byte[] payload = BinaryTool.toBytes(arguments);
        Request request = new Request();
        request.setType(0x8);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(arguments.getRid());
        return basicResponse.isOk();
    }

    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties,
                                byte[] body) throws IOException {
        BasicPublishArguments basicPublishArguments = new BasicPublishArguments();
        basicPublishArguments.setRid(createRidByUUID());
        basicPublishArguments.setChannelId(channelId);
        basicPublishArguments.setExchangeName(exchangeName);
        basicPublishArguments.setRoutingKey(routingKey);
        basicPublishArguments.setBasicProperties(basicProperties);
        basicPublishArguments.setBody(body);
        byte[] payload = BinaryTool.toBytes(basicPublishArguments);
        Request request = new Request();
        request.setType(0x9);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(basicPublishArguments.getRid());
        return basicResponse.isOk();
    }

    public boolean basicConsumer(String queueName , boolean autoAck, Consumer consumer) throws MqException, IOException {
        if(this.consumer != null) {
            throw new MqException("[Channel] 当前 channel 已经存在 consumer ！ channelId=" + channelId);
        }
        this.consumer = consumer;
        BasicConsumerArguments basicConsumerArguments = new BasicConsumerArguments();
        basicConsumerArguments.setRid(createRidByUUID());
        basicConsumerArguments.setChannelId(channelId);
        basicConsumerArguments.setConsumerTag(channelId);
        basicConsumerArguments.setQueueName(queueName);
        basicConsumerArguments.setAutoAck(autoAck);
        byte[] payload = BinaryTool.toBytes(basicConsumerArguments);
        Request request = new Request();
        request.setType(0xa);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(basicConsumerArguments.getRid());
        return basicResponse.isOk();
    }

    public boolean basicAck(String queueName, String messageId) throws IOException {
        BasicAckArguments arguments = new BasicAckArguments();
        arguments.setRid(createRidByUUID());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        arguments.setMessageId(messageId);
        byte[] payload = BinaryTool.toBytes(arguments);
        Request request = new Request();
        request.setType(0xb);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicResponse basicResponse = waitResponse(arguments.getRid());
        return basicResponse.isOk();
    }


}
